package com.stone.customer.console.trigger;

import com.stone.customer.core.domain.entity.Channel;
import com.stone.message.adapter.domain.bo.TriggerMessageParam;
import com.stone.message.adapter.infra.publisher.MessagePublisher;
import com.stone.message.adapter.trigger.AbstractCommonReceiverTrigger;
import com.stone.starter.core.constant.BaseConstants;
import com.stone.starter.redis.client.RedisCacheClient;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;

import java.util.ArrayList;
import java.util.List;
import java.util.StringJoiner;

/**
 * job同步Redis公用同步器
 *
 * @author Mr_wenpan@163.com 2021/07/17 16:40
 */
@Slf4j
@Deprecated
//@Component
public class JobSyncCommonTrigger extends AbstractCommonReceiverTrigger<List<Channel>, String> {

    /**
     * 门店收藏虚拟渠道添加到收藏夹
     */
    private static final String LUA_SCRIPT_VIRTUAL_CHANNEL_ADD = "script/lua/virtual_channel_get.lua";
    /**
     * 门店收藏虚拟渠道添加到收藏夹脚本
     */
    private static final DefaultRedisScript<List<Channel>> VIRTUAL_CHANNEL_GET_SCRIPT;

    /**
     * redis服务客户端
     */
    @Autowired
    private RedisCacheClient redisCacheClient;

    @Autowired
    @Qualifier("receiverFromRedisTriggerPool")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    static {
        VIRTUAL_CHANNEL_GET_SCRIPT = new DefaultRedisScript<>();
        VIRTUAL_CHANNEL_GET_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource(LUA_SCRIPT_VIRTUAL_CHANNEL_ADD)));
//        VIRTUAL_CHANNEL_GET_SCRIPT.setResultType(List.class);
    }

    public JobSyncCommonTrigger(MessagePublisher messagePublisher) {
        super(messagePublisher);
    }

    public void executeTask(String mqQueue, String redisQueue, Integer threadCount) {
        TriggerMessageParam<String> param = TriggerMessageParam.<String>builder()
                .queue(mqQueue)
                .queryDataParam(redisQueue)
                .taskCount(threadCount)
                .build();
        triggerWithCheck(param, this::compensate);
    }

    @Override
    protected ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        return threadPoolTaskExecutor;
    }

    @Override
    protected List<Channel> getData(TriggerMessageParam<String> param) {

        // job获取需要处理的数据
        // 通过lua脚本从Redis获取数据，获取成功后返回并删除zset数据
        List<String> keys = new ArrayList<>(1);
        // 会员收藏夹key KEY[1]
        keys.add("favoritesKey");
        Object[] args = new Object[2];
        // ARGV[1]
        args[0] = ("joinDate");
        // ARGV[2]
        StringJoiner joiner = new StringJoiner(BaseConstants.Symbol.LOWER_LINE);
        joiner.add("channel");
        joiner.add("posCode");
        args[1] = joiner.toString();
        return redisCacheClient.execute(VIRTUAL_CHANNEL_GET_SCRIPT, keys, args);
    }

    /**
     * 错误补偿方法
     *
     * @param data  需要补偿的数据
     * @param param 参数
     * @author Mr_wenpan@163.com 2021/7/17 2:07 下午
     */
    private void compensate(List<Channel> data, TriggerMessageParam<String> param) {
        // 写入数据库的错误记录表，记录出错数据、出错数据所对应的Redis队列的key，出错时间，出错类型，错误次数等
        // 出错类型: 有可能是多种错误情况写入同一张表，比如从Redis消费的错误，或job执行错误等
        // 错误次数: 方便多次重试还是出错的情况发送告警邮件，人工干预检查是否数据有问题
        String redisQueue = param.getQueryDataParam();
        String mqQueue = param.getQueue();

        // 获取job名称
        final String jobName = "shopSyncJob";

        log.error("job同步数据出错啦！jobName={}，redisQueue={},mqQueue={},data={}", jobName, redisQueue, mqQueue, data);

        // todo 错误信息写入数据库错误记录表

    }
}
